// *********************************************************************************************************
// *********************************************************************************************************
// Multiple Linear regression by gradient descent in Hadoop (many mappers-one reducer)
//
// Running
//     "bin/hadoop jar MultipleLinearRegressionMapReduce.jar com.ML_Hadoop.MultipleLinearRegression.MultipleLinearRegressionMapReduce ...
//                 <input_data_path> <output_path> <alpha, by default=0.1> <iteration, by default=100> ...
//                 <feature_size> <input_data_size>"
//     To stop confusion, it is recommended to remove "/user/hduser/LinearReg" before starting a new job by 
//     "bin/hadoop dfs -rmr "/user/hduser/LinearReg
//
// Inputs 
//     <input_data_path>  input text files/directory. Each line in the text file is as follows(https://gist.github.com/4202286):
//                        [yreal, feature-1, feature-2, ..., feature-N]
//                          151 , 0.038    , 0.05     , ..., 0.061 
//     alpha              The learning rate for gradient descent. Its value is between [0.001, 1]. By default its value is 0.1
//     iteration          The number of iterations. Its value is 100 by default.
//     feature_size       The number of features in input data. feature_size = (number of items in a line of input text file)-1.
//     input_data_size    The number of total lines in the input data. Values of "feature_size" and "input_data_size" are necessary 
//                        to upgrade regression coefficients (thata).
//
// Outputs 
//     "/user/hduser/LinearReg"  Keeps the history of values of regression coefficients (thata) for each iteration.
//                               Its format is [cost/error, theta-1, theta-2, ..., theta-N].
//
//     "/user/hduser/theta.txt"  Keeps the latest update of thata. Its format is [cost/error, theta-1, theta-2, ..., theta-N].
//
// Examples
//     "bin/hadoop jar MultipleLinearRegressionMapReduce.jar com.ML_Hadoop.MultipleLinearRegression.MultipleLinearRegressionMapReduce ...
//                 /user/hduser/diabetes /user/hduser/output-LR 0.1 3 11 442
//
// Author
//   Nikzad Babaii Rizvandi <nikzad.b(at)gmail.com>
//
// License
//   The program is free to use for non-commercial academic purposes,
//   but for course works, you must understand what is going inside to use.
//   The program can be used, modified, or re-distributed for any purposes
//   if you or one of your group understand codes. Please note that the author
//   does not guarantee the code works properly for all applications; therefore 
//   the author does not accept any responsibility on any damage/lost by using
//   this code.
//
// Changes
//   08/02/2013  First Edition
// *********************************************************************************************************
// *********************************************************************************************************

package cn.edu.xmu.datamining.tangzk.mralgos.linreg;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Joiner;

import cn.edu.xmu.datamining.tangzk.mralgos.AlgorithmDriver;
import cn.edu.xmu.datamining.tangzk.mralgos.kmeans.KMeansDriver;
import cn.edu.xmu.datamining.tangzk.util.StopWatch;

public class MultipleLinearRegressionMapReduce extends Configured implements
		Tool {

	private static final Logger LOG = LoggerFactory
			.getLogger(MultipleLinearRegressionMapReduce.class);

	public static void main(String[] args) throws Exception {
		AlgorithmDriver.toolRun(new MultipleLinearRegressionMapReduce(), args);
	}

	@Override
	public int run(String[] args) throws Exception {
		if (args.length < 6) {
			System.err
					.println("Usage: prog <inputDir> <outputDir> <alpha> <numIter> <feaSize> <dataSize>");
			return 1;
		}
		String[] theta;
		int iteration = 0, num_of_iteration = 1;
		int feature_size = 0, input_data_size = 0;
		FileSystem fs;
		Float alpha = 0.1f;
		StopWatch stopWatch = new StopWatch();
		do {
			Configuration conf = new Configuration();
			fs = FileSystem.get(conf);

			Job job = new Job(conf, "LinearRegressionMapReduce-"
					+ (iteration + 1));
			job.setJarByClass(MultipleLinearRegressionMapReduce.class);

			// the following two lines are needed for propagating "theta"
			conf = job.getConfiguration();

			job.setOutputKeyClass(LongWritable.class);
			job.setOutputValueClass(FloatWritable.class);

			job.setMapperClass(MultipleLinearRegressionMap.class);
			job.setReducerClass(MultipleLinearRegressionReduce.class);

			job.setInputFormatClass(TextInputFormat.class);
			job.setOutputFormatClass(TextOutputFormat.class);
			// only one reducer
			job.setNumReduceTasks(1);

			FileInputFormat.addInputPath(job, new Path(args[0]));
			String baseOutputDir = args[1];
			Path outputPath = new Path(getIterOutputDir(baseOutputDir,
					iteration));
			if (fs.exists(outputPath))
				fs.delete(outputPath, true);

			FileOutputFormat.setOutputPath(job, outputPath);
			alpha = Float.parseFloat(args[2]);
			num_of_iteration = Integer.parseInt(args[3]);
			feature_size = Integer.parseInt(args[4]);
			input_data_size = Integer.parseInt(args[5]);
			conf.setFloat("alpha", alpha);
			conf.setInt("feature_size", feature_size);
			conf.setInt("input_data_size", input_data_size);
			conf.setInt("iteration", iteration);

			theta = new String[feature_size];
			if (iteration == 0) { // first iteration
				for (int i = 0; i < theta.length; i++)
					theta[i] = "0.0";
			} else {
				try {
					String uri = getIterOutputDir(baseOutputDir, iteration - 1)
							+ "/theta.txt";
					fs = FileSystem.get(conf);
					// FSDataInputStream in = fs.open(new Path(uri));
					BufferedReader br = new BufferedReader(
							new InputStreamReader(fs.open(new Path(uri))));
					theta = br.readLine().split(",");
				} catch (Exception e) {

				}
			}
			conf.set("theta", Joiner.on(",").join(theta));
			LOG.info(String.format(
					"MapReduce Linear Regression weights: [%s]\n",
					Joiner.on(',').join(theta)));

			try {
				job.waitForCompletion(true);
				iteration++;
				Counters counters = job.getCounters();
				long totalCost = counters.findCounter("UserDefined",
						"TotalCost").getValue();
				long avgCost = counters.findCounter("UserDefined", "AvgCost")
						.getValue();
				LOG.info(String.format(
						"=== iteration %d, loss: %f, lossSum: %f. %fs===",
						iteration, avgCost / 1000.0f, totalCost / 1000.0f,
						stopWatch.elapsedTime()));
			} catch (IOException e) {
				e.printStackTrace();
			}
		} while (iteration < num_of_iteration);
		return 0;
	}

	public static String getIterOutputDir(String baseDir, int iteration) {
		return baseDir + "/linreg-iter-" + iteration;
	}
}